/*
 * Licensed to the Tai Ping Jin Ke
 *
 * Copyright (c) 2022 .
 * All rights reserved.
 * 项目名称：太平金科-基础组件封装层-ElasticSearch-缺省实现
 * 版权说明：本软件属太平金融科技服务(上海)有限公司所有，在未获得太平金融科技服务(上海)有限公司正式授权情况下，任何企业和个人，不能获取、阅读、安装、传播本软件涉及的任何受知识产权保护的内容。
 */
package com.taipingframework.boot.es.facade;

import com.taipingframework.utility.exception.UtilityException;
import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.admin.indices.get.GetIndexRequest;
import org.elasticsearch.common.settings.Settings;
import org.springframework.data.elasticsearch.client.reactive.ReactiveElasticsearchClient;
import reactor.core.publisher.Mono;

/**
 * 索引库管理 - 异步非阻塞
 */
public class ReactorIndexAdmin {

    private ReactiveElasticsearchClient reactiveElasticsearchClient;

    public ReactorIndexAdmin(ReactiveElasticsearchClient reactiveElasticsearchClient) {
        this.reactiveElasticsearchClient = reactiveElasticsearchClient;
    }

    /**
     * 索引是否存在
     */
    public Mono<Boolean> exists(String indexName) {
        return reactiveElasticsearchClient.indices()
                .existsIndex(new GetIndexRequest().indices(indexName))
                .onErrorResume(e -> Mono.error(new UtilityException(e)));
    }

    /**
     * 新增索引
     */
    public Mono<Boolean> create(String indexName, int shards, int replicas) {
        return Mono.defer(() ->
                Mono.fromSupplier(() -> {
                    Settings settings = Settings.builder()
                            // 分片
                            .put("number_of_shards", shards)
                            // 副本，单个节点的ES集群无法分配副本节点
                            .put("number_of_replicas", replicas)
                            .build();
                    CreateIndexRequest createIndexRequest = new CreateIndexRequest(indexName, Settings.Builder.EMPTY_SETTINGS);
                    createIndexRequest.index(indexName);
                    createIndexRequest.settings(settings);
                    return createIndexRequest;
                }))
                .flatMap(createIndexRequest -> reactiveElasticsearchClient.indices()
                        .createIndex(createIndexRequest)
                        .thenReturn(true)
                )
                .onErrorResume(e -> Mono.error(new UtilityException(e)));
    }

    /**
     * 删除索引
     */
    public Mono<Boolean> delete(String indexName) {
        return reactiveElasticsearchClient.indices()
                .deleteIndex(new DeleteIndexRequest(indexName))
                .thenReturn(true)
                .onErrorResume(e -> Mono.error(new UtilityException(e)));
    }

}
